using System;
using System.Collections.Generic;
using System.Linq;
using System.Text;
using MessageServer.channels;
using MessageServer.utilities;
using MessageServer.config;
using MessageServer.threadpool;
using MessageServer.logging;
using MessageServer.Handlers;
using System.Threading;
using MessageServer.persistence;

namespace MessageServer.channels.pipelines
{
    /**
     * This class is a task class that runs the handlers as an asynchronous task.
     * Assumption is made that the handlers are thread safe and no thread locking
     * is done here.
     * */
    class StreamTask
    {
        ChannelEvent _event;                    //channel event for which this task is run
        AsynchronousPipeline _pipe;             //The pipe to which this task belongs
        bool _streamup;                         //if the task should stream up or down
        int _task;                              //a unique task id assigned for tracking purpose

        /**
         * Constructor. This cannot be initialized outside this code module. Will be
         * used by the AsynchronousPipeline only.
         * */
        internal StreamTask(ChannelEvent evt, AsynchronousPipeline pipe, bool streamup, int task)
        {
            _event = evt;
            _pipe = pipe;
            _streamup = streamup;
            _task = task;
        }

        /**
         * The function that is run when the task is run. The task is queued into the
         * task thread pool and run when a thread becomes available.
         * */
        public void stream()
        {
            PerfStat s = new PerfStat();
            s.start("StreamTask:stream:" + _task);
            ChannelLogger.LOG.LogDebug("StreamTask:stream:starting Task:" + _task + ":" + _streamup + ":" + _event.GetType().Name);
            _pipe.asyncStream(_event, _streamup);
            Interlocked.Decrement(ref _pipe._tasks);
            ChannelLogger.LOG.LogDebug("StreamTask:stream:ending Task:" + _task);
            s.dumpperf(ChannelLogger.LOG);
        }

        public void handleException(IGatedTask tsk, Exception e)
        {
            ChannelLogger.LOG.LogDebug("StreamTask:handleException:ending:" + _task);
            ChannelLogger.LOG.LogError(e, e.Message);
            ChannelUtils.fireExceptionEvent(_event, e);
        }
    }

    /**
     * A streamable task stored in a queue. A dummy task is created to mark the start 
     * and end of the tasks queued in this pipeline.
     * */
    class StreamableTask
    {
        internal StreamTask _task;  //a null task indicating a dummy task, else the actual task

        /**
         * default constructor.
         * */
        internal StreamableTask()
        {
        }

        /**
         * if the task is null, this is considered as a dummy task and marks
         * the start and end.
         * */
        public bool isdummy()
        {
            return (_task == null);
        }
    }

    /**
     * A ChannelPipeline implementation class that runs handlers asynchronously. All handlers are
     * run within the same thread in the sequence they were registered, but the handlers for each 
     * event is run in a separate thread. This accepts an executor which decides how the handlers are
     * run, if the executor passed is a threadpool executor, then this runs in a thread pool, if an asynchronous
     * executor is passed, then this is run as separate thread without queueing.
     * This class derives from the defaultchannelpipeline and overrides the streamup and streamdown 
     * methods to run the base functionality in separate threads.
     * */
    class AsynchronousPipeline : DefaultChannelPipeline
    {
        Executor _asyncExecutor;                //Executor used to run asynchronous tasks
        bool _asynchStreamUp;                   //one time setup which indicates whether the streamup shd be synch or asynch
        bool _asynchStreamDown;                 //one time setup which indicates whether the streamdown shd be synch or not
        TSList<StreamableTask> _runStreams;     //the pending tasks to be run
        internal int _tasks;                             //the number of tasks run
        int _base;                              //a base number from which the tasks will start for this pipe

        /**
         * Construct a pipeline which executes asynch tasks using exec object and 
         * starts counting tasks from b.
         * */
        public AsynchronousPipeline(Executor exec, int b) : base()
        {
            _asyncExecutor = exec;
            if (exec == null)
                throw new Exception("Error, cannot execute on a null executor");
            //TODO: Will come back, need to test something here.
            _asynchStreamDown = false;
            _asynchStreamUp = true;
            _runStreams = new TSList<StreamableTask>(new StreamableTask());
            _base = b;
            _tasks = (b * 1000);
        }

        /**
         * Schedule the next task to be run. This is a callback function that will be called by the
         * thread pool once one task is run. This calls the next task to run and hence the stream of
         * tasks run is kept up till all tasks are executed.
         * */
        public void scheduleNext(IGatedTask t)
        {
            PerfStat s = new PerfStat();
            s.start("AsynchronousPipeline:scheduleNext:callback scheduling Next");
            //Note here, if it is decremeneted here, it is quite possible that this
            //has a clash with the executeTask function and the task is determined to
            //be queued rather than executed, and the pop here comes before the push there
            //this results in the loss of that job. Decrementing it before it reaches
            //here forces the executeTask to detect it as execute, only possibility there
            //is that more than one task of the same pipe is scheduled in the pool, which is
            //fine. So, decrementing this in the streamtask, which gives the executetask
            //more time to finish it's push.
            //Interlocked.Decrement(ref _tasks);
            ChannelLogger.LOG.LogDebug("AsynchronousPipeline:scheduleNext:Queueing Next");
            StreamableTask tsk = _runStreams.pop();
            if (!tsk.isdummy())
                _asyncExecutor.execute(tsk._task.stream, scheduleNext, tsk._task.handleException);
            s.dumpperf(ChannelLogger.LOG);
        }

        /**
         * A helper function to execute tasks. If there are no queued tasks and this is the
         * first task, then the task is directly assigned to the executor to execute, else
         * the tasks are queued into the runstreams to be called when this task completes.
         * */
        private void executeTask(StreamTask tsk)
        {
            if (_tasks == ((_base * 1000) + 1)) //first one
            {
                PerfStat s = new PerfStat();
                s.start("AsynchronousPipeline:executeTask:" + _tasks);
                _asyncExecutor.execute(tsk.stream, scheduleNext, tsk.handleException);
                ChannelLogger.LOG.LogDebug("AsynchronousPipeline:executeTask:Executing immediately:");
                s.dumpperf(ChannelLogger.LOG);
            }
            else
            {
                StreamableTask t = new StreamableTask();
                t._task = tsk;
                _runStreams.push(t);
                ChannelLogger.LOG.LogDebug("AsynchronousPipeline:executeTask:Queued");
            }
        }


        /**
         * The overridden streamup function from the base. This function creates a new
         * streamtask and calls the executetask function to execute the task. If asynch is not
         * required for stream up, then this executes it immediately.
         * */
        public override void  streamup(ChannelEvent evt)
        {
            //assumption is that the asynch executor takes care of queing if no thread available.
            //ensure only one is running per pipeline, so that we don't need too many of the threads in the
            //pool
            if (_asynchStreamUp)
            {
                Interlocked.Increment(ref _tasks);
                StreamTask tsk = new StreamTask(evt, this, true, _tasks);
                executeTask(tsk);
            }
            else
                asyncStream(evt, true);
        }

        /**
         * The overridden streamdown task that creates a streamtask for the streamdown task and 
         * calls the executetask to execute it if asynchronous task is required. Else this runs the
         * tasks synchronous with the called. Since stream down is called when an object is written to
         * the pipeline to be sent to the channel which is already in a separate thread, this function
         * will by default run synchronously.
         * */
        public override void  streamdown(ChannelEvent evt)
        {
            //assumption is that the asynch executor takes care of queing if no thread available.
            //ensure only one is running per pipeline, so that we don't need too many of the threads in the
            //pool
            if (_asynchStreamDown)
            {
                Interlocked.Increment(ref _tasks);
                StreamTask tsk = new StreamTask(evt, this, false, _tasks);
                executeTask(tsk);
            }
            else
                asyncStream(evt, false);
        }

        /**
         * The function called by the StreamTask class to execute
         * the base streamup and streamdown functions.
         * */
        internal void asyncStream(ChannelEvent evt, bool streamup)
        {
            if (streamup)
            {
                base.streamup(evt);
            }
            else
            {
                base.streamdown(evt);
            }
        }
    }

    /**
     * The Factory class used to create the AsynchronousPipeline class. This class creates a single threadpool
     * and uses this threadpool for all the pipelines creates. This creates a single threadpool to which 
     * all the pipelines can use to schedule tasks.s
     * */
    public class ThreadPoolPipelineFactory : PipeLineFactory
    {
        int _waittime;                  //the waittime for waiting for a thread in the pool
        int _min;                       //the minimum number of threads that will be created in the pool
        int _max;                       //the maximum number of threads that will be created in the pool
        String _name;                   //the name that will be used for the pool
        Executor _executor;             //the threadpool executor created
        Handler[] _handlers;            //the handlers for which the pipeline will be created
        int _created;                   //the number of pipelines created
        MessageReader _reader;          //the reader that should be used.
        TuningParams _parms;           //config
        MessageServerConfig _config;     //config

        /**
         * The constructor that is used to create the factory object.
         * */
        public ThreadPoolPipelineFactory(String nm, Handler[] handlers, 
            MessageReader reader, MessageServerConfig config, TuningParams parms)
        {
            _config = config;
            _parms = parms;
            _waittime = _parms.WaitForThread;
            _min = _parms.MinPipelineThreads;
            _max = _parms.MaxPipelineThreads;
            _name = nm;
            _reader = reader;

            //since this is a thread pool one everyone share the same pipeline
            _executor = new ThreadPoolExecutor(_name, _waittime, _min, _max);
            _handlers = handlers;
            _created = 0;
        }

        /**
         * A helper function used to create the pipeline.
         * */
        private ChannelPipeline createNew()
        {
            Interlocked.Increment(ref _created);
            ChannelPipeline pipe = new AsynchronousPipeline(_executor, _created);
            
            MessageFormatHandler read = new MessageFormatHandler();
            read.registerReader(_reader);
            pipe.addLast(read.handle, read); //register the first handler as the reader

            if (_config.persistentMessages)
            {
                MessagePersistenceHandler persist = new MessagePersistenceHandler(_config, _parms);
                pipe.addLast(persist.handle, persist);
            }

            MessageProcessorHandler process = new MessageProcessorHandler(_config, _parms);
            pipe.addLast(process.handle, process);

            if ((_handlers != null) && (_handlers.Length > 0))
            {
                for (int i = 0; i < _handlers.Length; i++)
                {
                    pipe.addLast(_handlers[i].handle, _handlers[i]);
                }
            }

            return pipe;
        }

        /**
         * The property used to create and access a new pipeline
         * */
        public ChannelPipeline pipeline { get { return createNew(); } }
    }
}
